/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.loading;

import java.nio.ByteBuffer;
import java.util.logging.Logger;

import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.storage.am.common.api.IIndexBulkLoadContext;
import edu.uci.ics.hyracks.storage.am.common.api.PageAllocationException;
import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.HyracksBtreeLexicon;

public class HyracksBTreeLexiconLoader extends HyracksBtreeLexicon implements
    ILexiconLoader {
  private static final float BTREE_FILL_FACTOR = 1.0f;
  private static final Logger LOG = Logger.getLogger(HyracksBTreeLexiconLoader.class
      .getName());

  private IIndexBulkLoadContext btreeBulkLoadCtx;
  private final FrameTupleAccessor btreeFrameTupleAccessor;
  private final FrameTupleReference btreeFrameTupleReference;
  private final FrameTupleAppender btreeTupleAppender;
  private final ByteBuffer btreeTupleBuffer;

  private final ArrayTupleBuilder btreeTupleBuilder;

  private boolean closed = false;

  public HyracksBTreeLexiconLoader() {
    int hyracksFrameSize = ColumnStore.BYTES_IN_DRAWER;
    RecordDescriptor recDesc =
        new RecordDescriptor(
                             new ISerializerDeserializer[] {
                                                            UTF8StringSerializerDeserializer.INSTANCE,
                                                            IntegerSerializerDeserializer.INSTANCE });

    this.btreeFrameTupleAccessor = new FrameTupleAccessor(hyracksFrameSize, recDesc);
    this.btreeTupleAppender = new FrameTupleAppender(hyracksFrameSize);
    this.btreeTupleBuffer = ByteBuffer.allocate(hyracksFrameSize);
    this.btreeTupleBuilder = new ArrayTupleBuilder(2);
    this.btreeFrameTupleReference = new FrameTupleReference();
  }

  @Override
  public void addKey(final String key, final int headerAddress) {
    try {
      // build tuple for inserting into btree
      this.btreeTupleBuilder.reset();
      this.btreeTupleBuilder
          .addField(UTF8StringSerializerDeserializer.INSTANCE, key);
      this.btreeTupleBuilder.addField(IntegerSerializerDeserializer.INSTANCE,
                                      headerAddress);

      // append tuple to temp buffer
      this.btreeTupleAppender.reset(this.btreeTupleBuffer, true);
      this.btreeTupleAppender.append(this.btreeTupleBuilder.getFieldEndOffsets(),
                                     this.btreeTupleBuilder.getByteArray(), 0,
                                     this.btreeTupleBuilder.getSize());

      // add tuple from temp buffer into btree. How clunky!
      this.btreeFrameTupleReference.reset(this.btreeFrameTupleAccessor, 0);
      this.lexicon.bulkLoadAddTuple(this.btreeFrameTupleReference,
                                    this.btreeBulkLoadCtx);

    } catch (HyracksDataException e) {
      throw new RuntimeException(e);
    } catch (PageAllocationException e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public void close() {
    if (this.closed) { return; }
    LOG.info("Closing: " + this.getClass().getName());
    try {
      this.lexicon.endBulkLoad(this.btreeBulkLoadCtx);
    } catch (HyracksDataException e) {
      e.printStackTrace();
      throw new RuntimeException(e);
    }
    super.close();
    this.closed = true;
  }

  public void initialize(final IBufferCache bufferCache, final String filename,
                         final IFileMapProvider fileMapProvider)
      throws HyracksDataException {
    super.initialize(bufferCache, filename, fileMapProvider, true);
    LOG.info("Initializing: " + this.getClass().getName());

    try {
      this.btreeBulkLoadCtx = this.lexicon.beginBulkLoad(BTREE_FILL_FACTOR);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }

    this.btreeFrameTupleAccessor.reset(this.btreeTupleBuffer);
  }

}
